package com.stormdemo.wclocalhost;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;

import com.stormdemo.wclocalhost.bolt.transfBolt;
import com.stormdemo.wclocalhost.bolt.wordCountBolt;
import com.stormdemo.wclocalhost.spout.worldInput;

public class ToupleMain {

	public static void main(String[] args) {
		TopologyBuilder touple = new TopologyBuilder();
		
		touple.setSpout("read", new worldInput(),2);
		touple.setBolt("transf", new transfBolt(),2).fieldsGrouping("read", new Fields("word"));
		touple.setBolt("wc",new wordCountBolt(),2).fieldsGrouping("transf", new Fields("transf"));
		
		
		Config conf = new Config();
		conf.setDebug(false);
		//conf.setMaxTaskParallelism(1);
		conf.put("filepath", "/home/read");
		//LocalCluster local = new LocalCluster();
		//local.submitTopology("wc", conf, touple.createTopology());
		try {
			StormSubmitter.submitTopology("wc", conf, touple.createTopology());
		} catch (AlreadyAliveException e) {
			
			e.printStackTrace();
		} catch (InvalidTopologyException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
